package com.zhangyang.esper.join;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.scopetest.SupportUpdateListener;
import com.espertech.esper.client.time.CurrentTimeEvent;
import com.espertech.esper.client.util.DateTime;
import com.espertech.esper.common.internal.support.SupportBean;
import com.espertech.esper.metrics.instrumentation.InstrumentationHelper;
import com.zhangyang.esper.support.bean.SupportBean_S5;
import com.zhangyang.esper.support.bean.SupportMarketDataBean;
import junit.framework.TestCase;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2022/7/28
 */
public class TestJoinOperator  extends TestCase {
    public EPServiceProvider epService;

    public  void testInnerJoinLastEvent(){

        Configuration config = new Configuration();
        config.getEngineDefaults().getLogging().setEnableQueryPlan(true);
        config.getEngineDefaults().getThreading().setInternalTimerEnabled(false);
        config.addEventType("A", SupportBean.class);
        config.addEventType("C", SupportBean_S5.class);
        config.addEventType("B", SupportMarketDataBean.class);
        epService = EPServiceProviderManager.getDefaultProvider(config);
        epService.initialize();
        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.startTest(epService, this.getClass(), getName());}

//        String stmtText = "select * from A.win:ext_timed(longPrimitive, 1 sec) as a  inner join B.win:ext_timed(longPrimitive, 1 sec) as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
//        String stmtText = "@Audit select * from A.win:length(1) as a  inner join B.win:length(1) as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
//        String stmtText = "@Audit select * from A.win:keepall() as a  inner join B.win:keepall() as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        String stmtText = "@Audit select * from A#lastevent as a  inner join B#lastevent as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        EPStatement statement = epService.getEPAdministrator().createEPL(stmtText);
        SupportUpdateListener listener = new SupportUpdateListener();
        statement.addListener(listener);

        // A(a=1) A(a=2) A(a=0) A(a=3) ...
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001"), "E1",1);
//        epService.getEPRuntime().sendEvent(new SupportBean("E1", 1));
//        epService.getEPRuntime().sendEvent(new SupportBean("E2", 2));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002"), "E2",2);
//        epService.getEPRuntime().sendEvent(new SupportBean("E3", 0));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.003")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.003"), "E3",0);
//        epService.getEPRuntime().sendEvent(new SupportBean("E5", 1));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.004")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.004"), "E5",1);
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.444")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.444"), "E5",1);
        SupportBean_S5 supportBean_s5 = new SupportBean_S5(1, "1");
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.500")));
        supportBean_s5.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:00.500"));
        epService.getEPRuntime().sendEvent(supportBean_s5);
//        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.000"), "E3",0);
        SupportMarketDataBean supportMarketDataBeanA = new SupportMarketDataBean("E6", "M1", 0);
        supportMarketDataBeanA.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanA);
        SupportMarketDataBean supportMarketDataBean = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBean.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBean);
        SupportMarketDataBean supportMarketDataBeanB = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBeanB.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanB);
        assertTrue(listener.isInvoked());

        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.endTest();}
    }

    public  void testInnerJoinLength(){

        Configuration config = new Configuration();
        config.getEngineDefaults().getLogging().setEnableQueryPlan(true);
        config.getEngineDefaults().getThreading().setInternalTimerEnabled(false);
        config.addEventType("A", SupportBean.class);
        config.addEventType("C", SupportBean_S5.class);
        config.addEventType("B", SupportMarketDataBean.class);
        epService = EPServiceProviderManager.getDefaultProvider(config);
        epService.initialize();
        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.startTest(epService, this.getClass(), getName());}

//        String stmtText = "select * from A.win:ext_timed(longPrimitive, 1 sec) as a  inner join B.win:ext_timed(longPrimitive, 1 sec) as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        String stmtText = "@Audit select * from A.win:length(10) as a  inner join B.win:length(10) as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
//        String stmtText = "@Audit select * from A.win:keepall() as a  inner join B.win:keepall() as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        EPStatement statement = epService.getEPAdministrator().createEPL(stmtText);
        SupportUpdateListener listener = new SupportUpdateListener();
        statement.addListener(listener);

        // A(a=1) A(a=2) A(a=0) A(a=3) ...
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001"), "E1",1);
//        epService.getEPRuntime().sendEvent(new SupportBean("E1", 1));
//        epService.getEPRuntime().sendEvent(new SupportBean("E2", 2));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002"), "E2",2);
//        epService.getEPRuntime().sendEvent(new SupportBean("E3", 0));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.003")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.003"), "E3",0);
//        epService.getEPRuntime().sendEvent(new SupportBean("E5", 1));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.004")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.004"), "E5",1);
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.444")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.444"), "E5",1);
        SupportBean_S5 supportBean_s5 = new SupportBean_S5(1, "1");
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.500")));
        supportBean_s5.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:00.500"));
        epService.getEPRuntime().sendEvent(supportBean_s5);
//        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.000"), "E3",0);
        SupportMarketDataBean supportMarketDataBeanA = new SupportMarketDataBean("E6", "M1", 0);
        supportMarketDataBeanA.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanA);
        SupportMarketDataBean supportMarketDataBean = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBean.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBean);
        SupportMarketDataBean supportMarketDataBeanB = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBeanB.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanB);
        //A和B流两者之间长度是10 ，10中两个流各有2个数据匹配，共匹配4对数据
        assertTrue(listener.isInvoked());

        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.endTest();}
    }

    public  void testInnerJoinTime(){

        Configuration config = new Configuration();
        config.getEngineDefaults().getLogging().setEnableQueryPlan(true);
        config.getEngineDefaults().getThreading().setInternalTimerEnabled(false);
        config.addEventType("A", SupportBean.class);
        config.addEventType("C", SupportBean_S5.class);
        config.addEventType("B", SupportMarketDataBean.class);
        epService = EPServiceProviderManager.getDefaultProvider(config);
        epService.initialize();
        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.startTest(epService, this.getClass(), getName());}

        String stmtText = "select * from A.win:ext_timed(longPrimitive, 1 sec) as a  inner join B.win:ext_timed(longPrimitive, 10 sec) as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        EPStatement statement = epService.getEPAdministrator().createEPL(stmtText);
        SupportUpdateListener listener = new SupportUpdateListener();
        statement.addListener(listener);

        // A(a=1) A(a=2) A(a=0) A(a=3) ...
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001"), "E1",1);
//        epService.getEPRuntime().sendEvent(new SupportBean("E1", 1));
//        epService.getEPRuntime().sendEvent(new SupportBean("E2", 2));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002"), "E2",2);
//        epService.getEPRuntime().sendEvent(new SupportBean("E3", 0));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.003")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.003"), "E3",0);
//        epService.getEPRuntime().sendEvent(new SupportBean("E5", 1));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.004")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.004"), "E5",1);
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:02.444")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:02.444"), "E5",1);
        SupportBean_S5 supportBean_s5 = new SupportBean_S5(1, "1");
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.500")));
        supportBean_s5.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:00.500"));
        epService.getEPRuntime().sendEvent(supportBean_s5);
//        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.000"), "E3",0);
        SupportMarketDataBean supportMarketDataBeanA = new SupportMarketDataBean("E6", "M1", 0);
        supportMarketDataBeanA.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanA);
        SupportMarketDataBean supportMarketDataBean = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBean.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBean);
        SupportMarketDataBean supportMarketDataBeanB = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBeanB.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanB);
        assertTrue(listener.isInvoked());

        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.endTest();}
    }

    protected void sendExtTimeEvent(long longPrimitive, String theString,int intPrimitive)
    {
        SupportBean theEvent = new SupportBean(theString, 0);
        theEvent.setLongPrimitive(longPrimitive);
        theEvent.setIntPrimitive(intPrimitive);
        epService.getEPRuntime().sendEvent(theEvent);
    }


    public  void testInnerJoinLengthByOrder(){

        Configuration config = new Configuration();
        config.getEngineDefaults().getLogging().setEnableQueryPlan(true);
        config.getEngineDefaults().getThreading().setInternalTimerEnabled(false);
        config.addEventType("A", SupportBean.class);
        config.addEventType("C", SupportBean_S5.class);
        config.addEventType("B", SupportMarketDataBean.class);
        epService = EPServiceProviderManager.getDefaultProvider(config);
        epService.initialize();
        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.startTest(epService, this.getClass(), getName());}

//        String stmtText = "select * from A.win:ext_timed(longPrimitive, 1 sec) as a  inner join B.win:ext_timed(longPrimitive, 1 sec) as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        String stmtText = "@Audit select * from A.win:length(1) as a  inner join B.win:length(1) as b on  a.theString = b.symbol ";
//        String stmtText = "@Audit select * from A.win:keepall() as a  inner join B.win:keepall() as b on  a.theString = b.symbol and a.intPrimitive = b.price ";
        EPStatement statement = epService.getEPAdministrator().createEPL(stmtText);
        SupportUpdateListener listener = new SupportUpdateListener();
        statement.addListener(listener);

        // A(a=1) A(a=2) A(a=0) A(a=3) ...
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.001"), "E6",1);
//        epService.getEPRuntime().sendEvent(new SupportBean("E1", 1));
//        epService.getEPRuntime().sendEvent(new SupportBean("E2", 2));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002")));
        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.002"), "E2",2);
//        epService.getEPRuntime().sendEvent(new SupportBean("E3", 0));




//        sendExtTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:00.000"), "E3",0);
        SupportMarketDataBean supportMarketDataBeanA = new SupportMarketDataBean("E6", "M1", 0);
        supportMarketDataBeanA.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanA);
        SupportMarketDataBean supportMarketDataBean = new SupportMarketDataBean("E5", "M1", 1);
        supportMarketDataBean.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:01.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBean);
        SupportMarketDataBean supportMarketDataBeanB = new SupportMarketDataBean("E2", "M1", 1);
        supportMarketDataBeanB.setLongPrimitive(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550"));
        epService.getEPRuntime().sendEvent(new CurrentTimeEvent(DateTime.parseDefaultMSec("2002-02-01T9:00:02.550")));
        epService.getEPRuntime().sendEvent(supportMarketDataBeanB);
        //A和B流两者之间长度是10 ，10中两个流各有2个数据匹配，共匹配4对数据
        assertTrue(listener.isInvoked());

        if (InstrumentationHelper.ENABLED) { InstrumentationHelper.endTest();}
    }
}
